package com.atguigu.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

object SparkStreaming07_Output1 {

    def main(args: Array[String]): Unit = {

        val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
        val ssc = new StreamingContext(sparkConf, Seconds(3))
        ssc.checkpoint("cp")

        val lines = ssc.socketTextStream("localhost", 9999)
        val wordToOne = lines.map((_, 1))

        val windowDS: DStream[(String, Int)] =
            wordToOne.reduceByKeyAndWindow(
                (x: Int, y: Int) => {
                    x + y
                },
                (x: Int, y: Int) => {
                    x - y
                },
                Seconds(9), Seconds(3))

        // foreachRDD不会出现时间戳
        //        注意：
        //        1)    连接不能写在 driver 层面(序列化)
        //        2)    如果写在 foreach 则每个 RDD 中的每一条数据都创建，得不偿失；
        //        3)    增加 foreachPartition ，在分区创建(获取) 。
        windowDS.foreachRDD(
            rdd => {

            }
        )

        ssc.start()
        ssc.awaitTermination()
    }

}
